package spcpy

import (
	"bytes"
	"context"
	"fmt"
	"gitee.com/ymofen/gobase"
	"gitee.com/ymofen/gobase/factory"
	"gitee.com/ymofen/gobase/getfrom"
	"gitee.com/ymofen/gobase/subpub"
	"gitee.com/ymofen/golog"
	"gitee.com/ymofen/sputils"
	"runtime"
	"sync"
	"sync/atomic"
	"time"
)

type LocalFileNode struct {
	startedflag int32
	taskId      string
	lk          sync.RWMutex
	confMap     gobase.StrMap
	ex          *subpub.Subchannel

	wg sync.WaitGroup

	preCheckBufVal interface{}
	checkBufFunc   func(buf []byte) interface{}

	interval time.Duration
	dataFrom string

	lastrecvT  time.Time
	statusText string
	cancelFunc func()
}

var (
	dataFromNodeAliveN int32
)

func GetLocalFileAliveN() int32 {
	return atomic.LoadInt32(&dataFromNodeAliveN)
}

func (this *LocalFileNode) Close() error {
	this.lk.Lock()
	fn := this.cancelFunc
	this.lk.Unlock()
	if fn != nil {
		fn()
		this.wg.Wait()
	}

	this.lk.Lock()
	this.cancelFunc = nil
	this.confMap = nil
	this.ex = nil
	this.startedflag = 0
	this.lk.Unlock()

	return nil
}

func (this *LocalFileNode) GetConnString() string {
	return ""
}

func (this *LocalFileNode) Sub(id, channel string, cb subpub.SubFunc) {
	this.ex.Sub(id, channel, cb)
}

func (this *LocalFileNode) Unsub(id, channel string) bool {
	return this.ex.Unsub(id, channel)
}

func (this *LocalFileNode) GetStatus(args ...interface{}) string {
	if this.startedflag == 0 {
		return fmt.Sprintf("服务已经关闭")
	}
	return this.statusText
}

func (this *LocalFileNode) doInterval() {
	dataFrom := gobase.ParseTimeFmt(this.dataFrom, time.Now())
	respBuf, err := getfrom.GetFromS(dataFrom)
	if err != nil {
		this.ex.Pub("", 0, sputils.SpTypeError, err)
	} else {
		chkFunc := this.checkBufFunc
		if chkFunc != nil {
			val := chkFunc(respBuf)
			if val == this.preCheckBufVal {
				return
			}
			this.preCheckBufVal = val
		}
		golog.Infof("%s, l:%d", dataFrom, len(respBuf))
		rd := bytes.NewReader(respBuf)
		block := make([]byte, 2048)
		for {
			n, e0 := rd.Read(block)
			if e0 != nil {
				break
			}
			if n > 0 {
				r0 := this.ex.Pub(sputils.SpDefaultChannel, 0, sputils.SpTypeData, block[:n], dataFrom)
				if r0 == -1 { // 推送错误
					break
				}
				time.Sleep(time.Millisecond)
				runtime.Gosched()
			}
		}

	}
}

func (this *LocalFileNode) CheckStart() error {
	this.lk.Lock()
	defer this.lk.Unlock()
	if this.startedflag == 1 {
		return nil
	}
	defer func() {
		this.startedflag = 1
	}()

	this.dataFrom = this.confMap.StringByName("datafrom", "")
	interval := gobase.ParseDurationEx(this.confMap.StringByName("interval", "300s"), 0)
	if interval < 0 {
		interval = time.Second * 10
	}
	this.interval = interval

	changedCheck := this.confMap.StringByName("checkChanged", "")
	if changedCheck == "crc" {
		this.checkBufFunc = func(buf []byte) interface{} {
			return gobase.CRC24QBuf(buf)
		}
	} else if changedCheck == "md5" {
		this.checkBufFunc = func(buf []byte) interface{} {
			return gobase.MD5Buf(buf)
		}
	}

	ctx, fn := context.WithCancel(context.Background())
	this.cancelFunc = fn
	this.wg.Add(1)
	go gobase.GoRun(func() {
		gobase.RunContextAndInterval(ctx, interval, func() {
			this.doInterval()
		})
		this.wg.Done()
	})

	return nil
}

var (
	DefaultIntervalFromNodes = gobase.NewConnectObjectRefStorage(newLocalFileNode)
)

func newLocalFileNode(connstr string) interface{} {
	node := &LocalFileNode{
		confMap: gobase.NewStrMap(),
		ex:      subpub.NewSubchannel(),
	}
	atomic.AddInt32(&dataFromNodeAliveN, 1)
	runtime.SetFinalizer(node, func(obj interface{}) {
		atomic.AddInt32(&dataFromNodeAliveN, -1)
	})
	node.confMap.URLFormDecode(connstr)

	return node
}

// conntype=datafrom&datafrom=file://&interval=30s
func checkGetDataFromReload(args ...interface{}) (rval interface{}, err error) {
	if len(args) == 0 {
		err = fmt.Errorf("无效的参数")
		return
	}
	if conf, ok := args[0].(gobase.StrMap); ok {
		s := conf.URLFormEncodeKeylst("datafrom", "interval", "checkChanged")
		rval = DefaultIntervalFromNodes.CheckGetRef(s)
		return
	} else {
		err = fmt.Errorf("无效的参数")
		return
	}
}

func init() {

	factory.RegisterCreateFunc("sp.datafrom", checkGetDataFromReload)
}
